Skip to main content

​
Introduction

The Tasks Module provides comprehensive functionality for managing task execution, monitoring progress, and streaming real-time events within the xpander.ai platform.

​
Overview

In this module, you can:
  • Create and execute tasks with detailed configuration options
  • Monitor task execution status and progress in real-time
  • Track custom internal status (up to 255 characters) alongside system status
  • Stream task events for responsive application development
  • Manage task lifecycle including stopping and updating tasks
  • Handle both synchronous and asynchronous task operations

​
Examples

​
Creating and Executing Tasks

This example demonstrates how to create and execute tasks with both asynchronous and synchronous approaches.

​
Asynchronous Example

Using acreate() to create tasks asynchronously: 
from xpander_sdk import Tasks

tasks = Tasks()
task = await tasks.acreate(
    agent_id="agent-123",
    prompt="Analyze the sales data",
    file_urls=["https://example.com/sales_data.csv"],
    events_streaming=True
)
print(f"Created task: {task.id}")

​
Synchronous Example

Using create() to create tasks synchronously: 
from xpander_sdk import Tasks

tasks = Tasks()
task = tasks.create(
    agent_id="agent-123",
    prompt="Analyze the sales data",
    file_urls=["https://example.com/sales_data.csv"]
)
print(f"Created task: {task.id}")

​
Retrieving Task Status

Learn how to retrieve and monitor task execution status.

​
Asynchronous Example

task = await tasks.aget("task-456")
print(f"Task status: {task.status}")
print(f"Task result: {task.result}")

​
Synchronous Example

task = tasks.get("task-456")
print(f"Task status: {task.status}")
print(f"Task result: {task.result}")

​
Streaming Task Events

Monitor real-time task execution with event streaming.

​
Asynchronous Example

async for event in task.aevents():
    print(f"Event: {event.type} at {event.time}")
    
    if event.type == "task_finished":
        print("Task completed!")
        break

​
Synchronous Example

for event in task.events():
    print(f"Event: {event.type}")
    
    if event.type == "task_finished":
        print("Task completed!")
        break

​
Managing Task Lifecycle

Control task execution with update and stop operations.

​
Asynchronous Example

# Stop a running task
stopped_task = await tasks.astop("task-456")
print(f"Task {stopped_task.id} has been stopped")

# Update task status
updated_task = await tasks.aupdate(
    task_id="task-456",
    status=AgentExecutionStatus.Completed,
    result="Analysis completed successfully"
)

​
Synchronous Example

# Stop a running task
stopped_task = tasks.stop("task-456")
print(f"Task {stopped_task.id} has been stopped")

# Update task status
updated_task = tasks.update(
    task_id="task-456",
    status=AgentExecutionStatus.Completed,
    result="Analysis completed successfully"
)

​
File Handling for Agno Integration

Demonstrate how to work with tasks that have file attachments and integrate them with Agno agents.

​
Working with Files and Images

# Create a task with various file types
task = await tasks.acreate(
    agent_id="agent-123",
    prompt="Analyze the attached documents and images",
    file_urls=[
        "https://example.com/report.pdf",      # PDF document
        "https://example.com/chart.png",       # Image file
        "https://example.com/data.csv",        # Human-readable file
        "https://example.com/logo.jpg"         # Another image
    ]
)

# Get files categorized by type for Agno integration
files = task.get_files()    # Returns Agno File objects for PDFs
images = task.get_images()  # Returns Agno Image objects for images
readable_files = task.get_human_readable_files()  # Returns content of text files

print(f"Found {len(files)} PDF files")
print(f"Found {len(images)} image files")
print(f"Found {len(readable_files)} readable files")

​
Agno Integration Example

from xpander_sdk import Backend
from agno.agent import Agent

# Initialize Agno agent with xpander backend
backend = Backend()
agno_agent = Agent(**backend.get_args())

# Pass files and images directly to Agno
result = await agno_agent.arun(
    input=task.to_message(),  # Includes text + file URLs + readable content
    files=files,              # PDF files as Agno File objects
    images=images            # Image files as Agno Image objects
)

print(f"Analysis result: {result.content}")
Continue to the [Tasks API Reference](/API reference/tasks/API reference) for detailed documentation on classes and methods.

Agents Module

Agent management and task creation

Events Module

Event-driven programming and real-time streaming

Tools Repository

Tool integration and invocation

Configuration

SDK configuration and authentication

​
Support

For additional help:
⌘I